package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderRefund {
    public static void main(String[] args) {
        //todo 1.环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //获取配置对象
        Configuration configuration = tenv.getConfig().getConfiguration();
        //为表关联时状态中储存的数据设置过期时间
        configuration.setString("table.exec.state.ttl","5s");
        // todo 2.状态后端设置

        //todo 3.从kafka读取topic_db数据,封装为Flink SQL表
       tenv.executeSql("create table topic_db("+
               "`database` string,"+
               "`table` string,"+
               "`type` string,"+
               "`data` map<string,string>,"+
               "`old` map<string,string>,"+
               "`proc_time` as PROCTIME(),"+
               "`ts` string "+")"+MyKafkaUtil.getKafkaDDL("topic_db","order_refund"));

       //todo 4.读取退单表数据
        Table orderRefundInfo = tenv.sqlQuery("select " +
                "data['id'] AS id, " +
                "data['user_id'] AS user_id, " +
                "data['order_id'] AS order_id, " +
                "data['sku_id'] AS sku_id, " +
                "data['refund_type'] AS refund_type, " +
                "data['refund_num'] AS refund_num, " +
                "data['refund_amount'] AS refund_amount, " +
                "data['refund_reason_type'] AS refund_reason_type, " +
                "data['refund_reason_txt'] AS refund_reason_txt, " +
                "data['create_time'] AS create_time, " +
                "proc_time, " +
                "ts " +
                "FROM topic_db " +
                "WHERE `table` = 'order_refund_info' AND `type` = 'insert'");
        tenv.createTemporaryView("order_refund_info",orderRefundInfo);

        //todo 5. 读取订单表数据，筛选退单数据
        Table orderInfoRefund = tenv.sqlQuery("select " +
                "data['id'] id," +
                "data['province_id'] province_id," +
                "`old`" +
                "from topic_db " +
                "where `table`='order_info' " +
                "and `type` ='update' " +
                "and data['order_status']='1005' " +
                "and `old` ['order_status'] is not null ");
        tenv.createTemporaryView("order_info_refund",orderInfoRefund);

        //todo 6.建立MySQL-LookUp 字典表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //todo 7.关联三张表获得退单宽表

        Table resultTable = tenv.sqlQuery("select " +
                " ri.id, " +
                " ri.user_id, " +
                " ri.order_id, " +
                " ri.sku_id, " +
                " oi.province_id, " +
                " date_format(ri.create_time,'yyyy-MM-dd') as date_id, " +
                " ri.create_time, " +
                " ri.refund_type, " +
                " type_dic.dic_name, " +
                " ri.refund_reason_type, " +
                " reason_dic.dic_name, " +
                " ri.refund_reason_txt, " +
                " ri.refund_num, " +
                " ri.refund_amount, " +
                " ri.ts, " +
                " current_row_timestamp() as row_op_ts " +
                " from `order_refund_info` ri" +
                " join " +
                "`order_info_refund` oi " +
                " on ri.order_id=oi.id " +
                " join " +
                " base_dic for system_time as of ri.proc_time as type_dic " +
                " on ri.refund_type=type_dic.dic_code " +
                " join " +
                " base_dic for system_time as of ri.proc_time as reason_dic " +
                " on ri.refund_reason_type=reason_dic.dic_code ");
        tenv.createTemporaryView("result_table",resultTable);


        //todo 8.建立Kafka-Connector dwd_trade_order_refund 表
        tenv.executeSql("create table dwd_trade_order_refund("+
        "id string,"+
        "user_id string,"+
        "order_id string,"+
        "sku_id string,"+
        "province_id string,"+
        "date_id string,"+
        "create_time string,"+
        "refund_type_code string,"+
        "refund_type_name string,"+
        "refund_reason_type_code string,"+
        "refund_reason_type_name string,"+
        "refund_reason_txt string,"+
        "refund_num string,"+
        "refund_amount string,"+
        "ts string,"+
        "row_op_ts timestamp_ltz(3)"+
        ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));


        //todo 9. 将关联结果写入 Kafka-Connector表
        tenv.executeSql("insert into dwd_trade_order_refund select * from result_table");

    }
}
